//! 实时通信模块集成测试
//! 包含WebSocket和SSE的完整端到端测试

use axum::{Router, extract::ws::{Message, WebSocket}, routing::get, response::sse::Event};
use axum_extra::headers::{authorization::Bearer, Authorization};
use axum_extra::TypedHeader;
use futures::{SinkExt, StreamExt};
use reqwest;
use serde_json::json;
use std::{net::SocketAddr, time::Duration};
use tokio::net::TcpStream;
use tokio_tungstenite::{connect_async, tungstenite::protocol::Message as WsMessage, MaybeTlsStream, WebSocketStream};

use crate::{
    realtime::{
        connection::{ConnectionManager, ConnectionManagerHandle},
        websocket::WebSocketHandler,
        sse::{SSEHandler, SSEMessageEndpoint},
        message::{HubMessage, HubResponse, MessageType, MessageSerializer},
    },
    services::{AuthConfig, AuthService, AuthServiceHandle, RealtimeAuthMiddleware},
    models::token_model::TokenModel,
};

/// 测试服务器配置
struct TestServer {
    addr: SocketAddr,
    connection_manager: ConnectionManagerHandle,
    auth_middleware: RealtimeAuthMiddleware,
}

impl TestServer {
    /// 创建测试服务器
    async fn new() -> Self {
        // 创建连接管理器
        let manager = ConnectionManager::new(100);
        let connection_manager = ConnectionManagerHandle::new(manager);
        
        // 创建认证服务
        let auth_config = AuthConfig::default();
        let auth_service = AuthService::new(auth_config).unwrap();
        let auth_service_handle = AuthServiceHandle::new(auth_service);
        let auth_middleware = RealtimeAuthMiddleware::new(auth_service_handle);
        
        // 创建WebSocket处理器
        let ws_handler = WebSocketHandler::new(connection_manager.clone(), auth_middleware.clone());
        
        // 创建SSE处理器
        let sse_handler = SSEHandler::new(connection_manager.clone(), auth_middleware.clone());
        let sse_endpoint = SSEMessageEndpoint::new(connection_manager.clone());
        
        // 创建路由
        let app = Router::new()
            .merge(ws_handler.create_route())
            .merge(sse_handler.create_route())
            .merge(sse_endpoint.message_route());
        
        // 启动服务器
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        
        tokio::spawn(async move {
            axum::serve(listener, app).await.unwrap();
        });
        
        // 等待服务器启动
        tokio::time::sleep(Duration::from_millis(500)).await;
        
        Self {
            addr,
            connection_manager,
            auth_middleware,
        }
    }
    
    /// 获取WebSocket URL
    fn ws_url(&self) -> String {
        format!("ws://{}/ws", self.addr)
    }
    
    /// 获取SSE URL
    fn sse_url(&self) -> String {
        format!("http://{}/sse", self.addr)
    }
    
    /// 获取SSE消息端点URL
    fn sse_message_url(&self) -> String {
        format!("http://{}/sse/message", self.addr)
    }
}

/// 创建测试Token
fn create_test_token() -> TokenModel {
    TokenModel {
        user_id: 12345,
        belong_id: 1,
        // 其他字段根据实际需要添加
    }
}

/// 生成测试Token字符串
fn generate_test_token() -> String {
    // 创建一个简单的测试Token（在实际项目中应该使用AuthService生成）
    // 这里我们使用一个简单的格式：user_id:belong_id:timestamp
    let timestamp = chrono::Utc::now().timestamp();
    format!("test_token_{}_{}_{}", 12345, 1, timestamp)
}

/// WebSocket连接测试客户端
struct WebSocketTestClient {
    stream: WebSocketStream<MaybeTlsStream<TcpStream>>,
}

/// WebSocket测试客户端错误类型
#[derive(Debug)]
pub enum WebSocketTestClientError {
    ConnectionError(String),
    MessageError(String),
}

impl std::fmt::Display for WebSocketTestClientError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            WebSocketTestClientError::ConnectionError(msg) => write!(f, "连接错误: {}", msg),
            WebSocketTestClientError::MessageError(msg) => write!(f, "消息错误: {}", msg),
        }
    }
}

impl std::error::Error for WebSocketTestClientError {}

impl WebSocketTestClient {
    /// 连接到WebSocket服务器（不带Token）
    async fn connect(url: &str) -> Result<Self, WebSocketTestClientError> {
        let (stream, _) = connect_async(url).await
            .map_err(|e| WebSocketTestClientError::ConnectionError(e.to_string()))?;
        Ok(Self { stream })
    }
    
    /// 连接到WebSocket服务器（带Token）
    async fn connect_with_token(url: &str, token: &str) -> Result<Self, WebSocketTestClientError> {
        let ws_url = format!("{}?access_token={}", url, token);
        let (stream, _) = connect_async(&ws_url).await
            .map_err(|e| WebSocketTestClientError::ConnectionError(e.to_string()))?;
        Ok(Self { stream })
    }
    
    /// 发送消息
    async fn send_message(&mut self, message: &str) -> Result<(), WebSocketTestClientError> {
        self.stream.send(WsMessage::Text(message.to_string())).await
            .map_err(|e| WebSocketTestClientError::MessageError(e.to_string()))?;
        Ok(())
    }
    
    /// 接收消息
    async fn receive_message(&mut self) -> Result<String, WebSocketTestClientError> {
        match self.stream.next().await {
            Some(Ok(WsMessage::Text(text))) => Ok(text),
            Some(Ok(WsMessage::Close(_))) => Err(WebSocketTestClientError::ConnectionError("连接关闭".to_string())),
            Some(Err(e)) => Err(WebSocketTestClientError::MessageError(e.to_string())),
            None => Err(WebSocketTestClientError::ConnectionError("连接断开".to_string())),
            _ => Err(WebSocketTestClientError::MessageError("收到非文本消息".to_string())),
        }
    }
    
    /// 发送Ping消息
    async fn send_ping(&mut self) -> Result<(), WebSocketTestClientError> {
        self.stream.send(WsMessage::Ping(vec![])).await
            .map_err(|e| WebSocketTestClientError::MessageError(e.to_string()))?;
        Ok(())
    }
    
    /// 关闭连接
    async fn close(&mut self) -> Result<(), WebSocketTestClientError> {
        self.stream.close(None).await
            .map_err(|e| WebSocketTestClientError::ConnectionError(e.to_string()))?;
        Ok(())
    }
}

#[cfg(test)]
mod integration_tests {
    use super::*;
    
    #[tokio::test]
    async fn test_websocket_connection_establishment() {
        // 创建测试服务器
        let server = TestServer::new().await;
        
        // 生成测试Token
        let test_token = generate_test_token();
        
        // 连接到WebSocket服务器（带Token）
        let mut client = WebSocketTestClient::connect_with_token(&server.ws_url(), &test_token).await.unwrap();
        
        // 等待连接完全建立
        tokio::time::sleep(Duration::from_millis(100)).await;
        
        // 接收连接确认消息
        let connect_response = client.receive_message().await.unwrap();
        let response: HubResponse = serde_json::from_str(&connect_response).unwrap();
        
        assert!(response.result.is_some());
        
        // 验证连接ID
        let result = response.result.unwrap();
        let connection_id = result["connectionId"].as_str().unwrap();
        assert!(!connection_id.is_empty());
        
        // 验证协议版本
        let protocol = result["protocol"].as_str().unwrap();
        assert_eq!(protocol, "signalr");
        
        let version = result["version"].as_u64().unwrap();
        assert_eq!(version, 1);
        
        println!("WebSocket连接建立测试通过");
    }
    
    #[tokio::test]
    async fn test_websocket_ping_pong() {
        let server = TestServer::new().await;
        let test_token = generate_test_token();
        let mut client = WebSocketTestClient::connect_with_token(&server.ws_url(), &test_token).await.unwrap();
        
        // 等待连接完全建立
        tokio::time::sleep(Duration::from_millis(100)).await;
        
        // 接收连接确认消息
        let _ = client.receive_message().await.unwrap();
        
        // 发送应用级别的Ping消息
        let ping_message = HubMessage {
            message_type: MessageType::Ping,
            invocation_id: Some("test-ping".to_string()),
            target: None,
            arguments: None,
            stream_ids: None,
            error: None,
            result: None,
            allow_reconnect: None,
            headers: None,
        };
        
        let ping_json = MessageSerializer::serialize(&ping_message).unwrap();
        client.send_message(&ping_json).await.unwrap();
        
        // 接收应用级别的Ping响应
        let pong_response = client.receive_message().await.unwrap();
        println!("收到的Ping响应: {}", pong_response);
        
        let response: HubResponse = serde_json::from_str(&pong_response).unwrap();
        
        println!("解析后的响应: invocation_id={:?}, result={:?}", response.invocation_id, response.result);
        
        assert_eq!(response.invocation_id, Some("test-ping".to_string()));
        // 注意：serde_json::Value::Null 序列化为 JSON 的 null，反序列化后为 None
        // 所以这里应该检查 result 是否为 None，而不是 Some(serde_json::Value::Null)
        assert_eq!(response.result, None);
        
        println!("WebSocket Ping-Pong测试通过");
    }
    
    #[tokio::test]
    async fn test_websocket_token_authentication() {
        let server = TestServer::new().await;
        let test_token = generate_test_token();
        let mut client = WebSocketTestClient::connect_with_token(&server.ws_url(), &test_token).await.unwrap();
        
        // 等待连接完全建立
        tokio::time::sleep(Duration::from_millis(100)).await;
        
        // 接收连接确认消息
        let _ = client.receive_message().await.unwrap();
        
        // 创建测试Token（使用TokenModel）
        let test_token_model = create_test_token();
        
        // 发送认证消息
        let auth_message = HubMessage {
            message_type: MessageType::Invocation,
            invocation_id: Some("auth-1".to_string()),
            target: Some("TokenAuth".to_string()),
            arguments: Some(vec![json!({
                "user_id": test_token_model.user_id,
                "belong_id": test_token_model.belong_id
            })]),
            stream_ids: None,
            error: None,
            result: None,
            allow_reconnect: None,
            headers: None,
        };
        
        let auth_json = MessageSerializer::serialize(&auth_message).unwrap();
        client.send_message(&auth_json).await.unwrap();
        
        // 接收认证响应
        let auth_response = client.receive_message().await.unwrap();
        println!("收到的Token认证响应: {}", auth_response);
        let response: HubResponse = serde_json::from_str(&auth_response).unwrap();
        println!("解析后的Token认证响应: invocation_id={:?}, result={:?}", response.invocation_id, response.result);
        
        assert_eq!(response.invocation_id, Some("auth-1".to_string()));
        
        // 检查result是否存在
        if let Some(result) = response.result {
            let success = result["success"].as_bool().unwrap();
            assert!(success);
        } else {
            panic!("Token认证响应缺少result字段");
        }
        
        println!("WebSocket Token认证测试通过");
    }
    
    #[tokio::test]
    async fn test_websocket_message_broadcast() {
        let server = TestServer::new().await;
        
        // 生成测试Token
        let test_token1 = generate_test_token();
        let test_token2 = generate_test_token();
        
        // 创建多个客户端连接
        let mut client1 = WebSocketTestClient::connect_with_token(&server.ws_url(), &test_token1).await.unwrap();
        let mut client2 = WebSocketTestClient::connect_with_token(&server.ws_url(), &test_token2).await.unwrap();
        
        // 等待连接完全建立
        tokio::time::sleep(Duration::from_millis(100)).await;
        
        // 接收连接确认消息
        let _ = client1.receive_message().await.unwrap();
        let _ = client2.receive_message().await.unwrap();
        
        // 广播消息
        let broadcast_message = HubResponse::success(
            None,
            json!({ "type": "broadcast", "data": "Hello, World!" }),
        );
        
        server.connection_manager.broadcast(broadcast_message).await;
        
        // 两个客户端都应该收到广播消息
        let msg1 = client1.receive_message().await.unwrap();
        let msg2 = client2.receive_message().await.unwrap();
        
        let response1: HubResponse = serde_json::from_str(&msg1).unwrap();
        let response2: HubResponse = serde_json::from_str(&msg2).unwrap();
        
        let result1 = response1.result.unwrap();
        let result2 = response2.result.unwrap();
        
        assert_eq!(result1["type"], "broadcast");
        assert_eq!(result2["type"], "broadcast");
        
        println!("WebSocket消息广播测试通过");
    }
    
    #[tokio::test]
    async fn test_sse_connection_establishment() {
        let server = TestServer::new().await;
        
        // 生成测试Token
        let test_token = generate_test_token();
        
        // 连接到SSE服务器
        let client = reqwest::Client::new();
        let response = client
            .get(&server.sse_url())
            .header("Authorization", format!("Bearer {}", test_token))
            .send()
            .await
            .unwrap();
        
        assert!(response.status().is_success());
        assert_eq!(response.headers().get("content-type").unwrap(), "text/event-stream");
        
        println!("SSE连接建立测试通过");
    }
    
    #[tokio::test]
    async fn test_sse_message_broadcast() {
        let server = TestServer::new().await;
        
        // 通过SSE消息端点发送广播消息
        let client = reqwest::Client::new();
        
        let broadcast_message = HubMessage {
            message_type: MessageType::Invocation,
            invocation_id: Some("broadcast-1".to_string()),
            target: Some("BroadcastMessage".to_string()),
            arguments: Some(vec![json!({ "message": "Broadcast via SSE" })]),
            stream_ids: None,
            error: None,
            result: None,
            allow_reconnect: None,
            headers: None,
        };
        
        let message_json = MessageSerializer::serialize(&broadcast_message).unwrap();
        
        let response = client
            .post(&server.sse_message_url())
            .body(message_json)
            .send()
            .await
            .unwrap();
        
        assert!(response.status().is_success());
        
        let result_text = response.text().await.unwrap();
        let result: serde_json::Value = serde_json::from_str(&result_text).unwrap();
        assert!(result["success"].as_bool().unwrap());
        
        println!("SSE消息广播测试通过");
    }
    
    #[tokio::test]
    async fn test_websocket_connection_cleanup() {
        let server = TestServer::new().await;
        
        // 获取初始连接数
        let initial_count = server.connection_manager.active_connections().await;
        
        // 生成测试Token
        let test_token = generate_test_token();
        
        // 创建连接
        let mut client = WebSocketTestClient::connect_with_token(&server.ws_url(), &test_token).await.unwrap();
        
        // 接收连接确认消息
        let _ = client.receive_message().await.unwrap();
        
        // 验证连接数增加
        let after_connect_count = server.connection_manager.active_connections().await;
        assert_eq!(after_connect_count, initial_count + 1);
        
        // 关闭连接
        client.close().await.unwrap();
        
        // 等待连接清理
        tokio::time::sleep(Duration::from_millis(500)).await;
        
        // 验证连接数恢复
        let after_close_count = server.connection_manager.active_connections().await;
        assert_eq!(after_close_count, initial_count);
        
        println!("WebSocket连接清理测试通过");
    }
    
    #[tokio::test]
    async fn test_websocket_concurrent_connections() {
        let server = TestServer::new().await;
        
        // 创建多个并发连接
        let mut handles = vec![];
        
        for i in 0..3 {
            let url = server.ws_url().clone();
            let handle = tokio::spawn(async move {
                // 增加连接之间的延迟，避免资源竞争
                tokio::time::sleep(Duration::from_millis(i * 50)).await;
                
                // 生成测试Token
                let test_token = generate_test_token();
                
                let mut client = WebSocketTestClient::connect_with_token(&url, &test_token).await.unwrap();
                
                // 等待连接完全建立
                tokio::time::sleep(Duration::from_millis(200)).await;
                
                // 接收连接确认消息（改进错误处理）
                match client.receive_message().await {
                    Ok(response) => {
                        let hub_response: HubResponse = serde_json::from_str(&response).unwrap();
                        assert!(hub_response.result.is_some());
                        
                        // 发送Ping消息
                        let ping_message = HubMessage {
                            message_type: MessageType::Ping,
                            invocation_id: Some(format!("ping-{}", i)),
                            target: None,
                            arguments: None,
                            stream_ids: None,
                            error: None,
                            result: None,
                            allow_reconnect: None,
                            headers: None,
                        };
                        
                        let ping_json = MessageSerializer::serialize(&ping_message).unwrap();
                        client.send_message(&ping_json).await.unwrap();
                        
                        // 接收响应
                        match client.receive_message().await {
                            Ok(pong_response) => {
                                let pong_hub_response: HubResponse = serde_json::from_str(&pong_response).unwrap();
                                assert!(pong_hub_response.result.is_some());
                            }
                            Err(e) => {
                                eprintln!("接收Pong响应失败: {}", e);
                                // 不panic，继续执行关闭操作
                            }
                        }
                    }
                    Err(e) => {
                        eprintln!("接收连接确认消息失败: {}", e);
                        // 不panic，继续执行关闭操作
                    }
                }
                
                // 安全关闭连接
                let _ = client.close().await;
            });
            
            handles.push(handle);
        }
        
        // 等待所有连接完成
        for handle in handles {
            // 不panic，允许连接失败
            let _ = handle.await;
        }
        
        println!("WebSocket并发连接测试完成");
    }
    
    #[tokio::test]
    async fn test_websocket_error_handling() {
        let server = TestServer::new().await;
        let test_token = generate_test_token();
        let mut client = WebSocketTestClient::connect_with_token(&server.ws_url(), &test_token).await.unwrap();
        
        // 等待连接完全建立
        tokio::time::sleep(Duration::from_millis(100)).await;
        
        // 接收连接确认消息
        let _ = client.receive_message().await.unwrap();
        
        // 发送无效消息格式
        client.send_message("invalid json").await.unwrap();
        
        // 应该收到错误响应
        let error_response = client.receive_message().await.unwrap();
        let response: HubResponse = serde_json::from_str(&error_response).unwrap();
        
        assert!(response.error.is_some());
        
        println!("WebSocket错误处理测试通过");
    }
}